package client.kafka;

import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.ResourceBundle;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.admin.TopicListing;
import org.apache.log4j.Logger;

import client.kafka.impl.DO.KafkaConsumerGroupListTreeNode;
import client.kafka.impl.DO.KafkaConsumerGroupTreeNode;
import client.kafka.impl.DO.KafkaRootTreeNode;
import client.kafka.impl.DO.KafkaTopicListTreeNode;
import client.kafka.impl.DO.KafkaTopicTreeNode;
import core.AlertUtil;
import core.FormDialog;
import core.TextAreaAppender;
import javafx.application.Platform;
import javafx.event.ActionEvent;
import javafx.fxml.FXML;
import javafx.fxml.Initializable;
import javafx.scene.control.ContextMenu;
import javafx.scene.control.MenuItem;
import javafx.scene.control.TextArea;
import javafx.scene.control.TextField;
import javafx.scene.control.TreeItem;
import javafx.scene.control.TreeView;
import javafx.scene.layout.BorderPane;
import javafx.stage.Stage;

public class KafkaClientController implements Initializable{

	/**
	 * Kafka客户端名称与实例的Map集合
	 */
	private static final Map<String,IKafkaClient> KAFKA_CLIENT=new HashMap<String,IKafkaClient>();
	static {
		//通过服务发现查找Kafka客户端实例
		ServiceLoader<IKafkaClient> clients=ServiceLoader.load(IKafkaClient.class);
		for(IKafkaClient client:clients) {
			KAFKA_CLIENT.put(client.getName(), client);
		}
	}

	//应用日志对象
	private Logger appLogger;

	@FXML
	private BorderPane container;

	/**
	 * 左侧树结构
	 */
	@FXML
	private TreeView<KafkaTreeNode> kafkaTreeView;

	/**
	 * 信息组件
	 */
	@FXML
	private TextArea infoTextArea;

	/**
	 * 命令历史组件
	 */
	@FXML
	private TextArea cmdTextArea;

	/**
	 * 命令输入框
	 */
	@FXML
	private TextField cmdTextField;

	/**
	 * 日志组件
	 */
	@FXML
	private TextArea logTextArea;

	/**
	 * Kafka客户端连接实例
	 */
	private IKafkaClient kafkaClient;

	@Override
	public void initialize(URL location, ResourceBundle resources) {
		appLogger = Logger.getLogger("kafka_client_app");
		appLogger.addAppender(new TextAreaAppender(logTextArea));
		appLogger.info("日志输出初始化完成！");

		Platform.runLater(()->{
			Stage stage = (Stage)container.getScene().getWindow();
			 stage.setOnCloseRequest((event)->{
				 if(kafkaClient!=null){
					 try {
						 kafkaClient.close();
					} catch (IOException e) {
						e.printStackTrace();
					}
				 }
			 });
		});
	}

	/**
	 * 选择Kafka客户端。
	 * 如果只有一个实例则直接使用，多个时弹出选择对话框进行选择。
	 * @return
	 */
	private IKafkaClient choiceClient() {
		appLogger.info("进行客户端选择。");
		if(KAFKA_CLIENT.size() == 1) {
			appLogger.info("仅有一个客户端示例，无需选择！");
			//只有一种客户端
			for(IKafkaClient client:KAFKA_CLIENT.values()) {
				return client;
			}
		}

		appLogger.info("启动客户端选择对话框。");
		//弹出选择对话框进行选择
		Set<String> clients = KAFKA_CLIENT.keySet();
		Optional<String> result = AlertUtil.choice("选择客户端", "请选择连接客户端类型：", clients);
		if(result.isPresent()) {
			appLogger.info("用户选择客户端："+result.get());
			return KAFKA_CLIENT.get(result.get());
		}

		return null;
	}

	/**
	 * 新建连接按钮监听
	 * @param e	事件对象
	 * @throws IOException　Kafka连接操作可能发生异常
	 */
	public void createConnection(ActionEvent e) throws IOException{
		appLogger.info("检查是否存在已连接的客户端。");
		//如果Kafka客户端实例不为空则先关闭之前的连接
		if(kafkaClient != null) {
			appLogger.info("关闭已连接的客户端。");
			kafkaClient.close();
		}

		//选择Kafka客户端
		this.kafkaClient = this.choiceClient();
		if(kafkaClient != null) {
			//初始化Kafka连接和组件
			if(kafkaClient.init()){
				cmdTextArea.setText("已连接！\n");
				initTreeView();
			}
		}
	}

	/**
	 * 初始化TreeView树
	 */
	private void initTreeView() {
		appLogger.info("初始化Kafka树结构。");
		TreeItem<KafkaTreeNode> root = this.getRootTreeItem();
		kafkaTreeView.setRoot(root);

		appLogger.info("添加节点选择监听事件！");
		//节点选择事件处理
		kafkaTreeView.getSelectionModel().selectedItemProperty().addListener((observable,oldValue,newValue)->{
			appLogger.info("Kafka树节点被点击 -> oldValue:"+oldValue+",newValue:"+newValue);
			TreeItem<KafkaTreeNode> currentSelectedItem = (TreeItem<KafkaTreeNode>) newValue;
			if(currentSelectedItem==null) return;

			//设置节点信息
			KafkaTreeNode selectedNode = currentSelectedItem.getValue();
			String itemInfo = this.getItemInfo(selectedNode);
			infoTextArea.setText(itemInfo);

			if(!selectedNode.isLeaf() && !selectedNode.isLoaded()) {
				currentSelectedItem.getChildren().clear();
				List<TreeItem<KafkaTreeNode>> childNodeList = this.getChildren(selectedNode);
				currentSelectedItem.getChildren().addAll(childNodeList);
				currentSelectedItem.setExpanded(true);
			}

			selectedNode.setLoaded(true);
		});

		appLogger.info("设置树节点右键菜单！");
		//设置TreeView树邮件菜单
		MenuItem refreshItem = new MenuItem("刷新");
		refreshItem.setOnAction((event)->{
			appLogger.info("执行[刷新]动作！");
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();

			//刷新信息
			infoTextArea.setText(this.getItemInfo(kafkaTreeNode));

			//刷新子节点
			if(!kafkaTreeNode.isLeaf()) {
				selectedItem.getChildren().clear();
				selectedItem.getChildren().addAll(this.getChildren(kafkaTreeNode));
				selectedItem.setExpanded(true);
			}
		});

		MenuItem disconnectItem = new MenuItem("断开连接");
		disconnectItem.setOnAction((event)->{
			appLogger.info("执行[断开连接]动作！");
			try {
				kafkaClient.close();
				kafkaTreeView.setRoot(null);
				kafkaTreeView.setContextMenu(null);
				infoTextArea.clear();
				cmdTextArea.setText("连接已断开！\n");
			} catch (IOException e) {
				AlertUtil.exception(e);
			}
		});

		MenuItem newTopicItem = new MenuItem("新建主题");
		newTopicItem.setOnAction((event)->{
			appLogger.info("执行[新建主题]动作！");
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			TextField nameTextField = new TextField();
			TextField partitionTextField = new TextField("1");
			TextField replicationTextField = new TextField("1");
			FormDialog newTopicDialog = new FormDialog("新建主题",
					new FormDialog.Item("主题名称", nameTextField),
					new FormDialog.Item("分区数", partitionTextField),
					new FormDialog.Item("拷贝数", replicationTextField));
			Optional<Boolean> result = newTopicDialog.showAndWait();
			if(result.isPresent()) {
				String topicName = nameTextField.getText();
				int numPartitions = Integer.valueOf(partitionTextField.getText());
				short replicationFactor = Short.valueOf(replicationTextField.getText());
				NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
				kafkaClient.createTopics(Arrays.asList(newTopic));

				//刷新子节点
				selectedItem.getChildren().clear();
				List<TreeItem<KafkaTreeNode>> childNodeList = this.getChildren(kafkaTreeNode);
				selectedItem.getChildren().addAll(childNodeList);
				selectedItem.setExpanded(true);
			}
		});

		MenuItem delTopicItem = new MenuItem("删除主题");
		delTopicItem.setOnAction((event)->{
			appLogger.info("执行[新建主题]动作！");
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			if(kafkaTreeNode instanceof KafkaTopicTreeNode) {
				String topic = ((KafkaTopicTreeNode) kafkaTreeNode).getTopic();
				kafkaClient.deleteTopics(Arrays.asList(topic));
				selectedItem.getParent().getChildren().clear(); //清理当前层级节点，重新加载
			}
		});

		kafkaTreeView.setContextMenu(new ContextMenu(disconnectItem));
		kafkaTreeView.setOnContextMenuRequested((event)->{
			kafkaTreeView.getContextMenu().getItems().clear();
			TreeItem<KafkaTreeNode> selectedItem = kafkaTreeView.getSelectionModel().getSelectedItem();
			KafkaTreeNode kafkaTreeNode = selectedItem.getValue();
			kafkaTreeView.getContextMenu().getItems().add(refreshItem);
			if(kafkaTreeNode instanceof KafkaRootTreeNode) {
				kafkaTreeView.getContextMenu().getItems().addAll(newTopicItem,disconnectItem);
			}else if(kafkaTreeNode instanceof KafkaTopicTreeNode) {
				kafkaTreeView.getContextMenu().getItems().add(delTopicItem);
			}
		});

		appLogger.info("Kafka树结构初始化完成！");
	}

	/**
	 * 获取根节点
	 * @return　根节点对象
	 */
	private TreeItem<KafkaTreeNode> getRootTreeItem() {
		appLogger.info("获取Kafka树根节点！");
		KafkaRootTreeNode rootNode = new KafkaRootTreeNode(kafkaClient.getConnStr());
		TreeItem<KafkaTreeNode> rootItem = new TreeItem<KafkaTreeNode>(rootNode);
		rootItem.setExpanded(true);
		rootItem.getChildren().addAll(this.getChildren(rootNode));
		return rootItem;
	}

	/**
	 * 获取子节点集合
	 * @param selectedNode　当前选中节点
	 * @return　子节点集合
	 */
	private List<TreeItem<KafkaTreeNode>> getChildren(KafkaTreeNode selectedNode) {
		appLogger.info("获取Kafka树子节点！当前节点："+selectedNode);
		if(selectedNode instanceof KafkaRootTreeNode) {
			List<TreeItem<KafkaTreeNode>> childrenList = new ArrayList<TreeItem<KafkaTreeNode>>();
			childrenList.add(new TreeItem<KafkaTreeNode>(new KafkaTopicListTreeNode())); //主题节点
			childrenList.add(new TreeItem<KafkaTreeNode>(new KafkaConsumerGroupListTreeNode())); //消费者节点
			return childrenList;
		}else if(selectedNode instanceof KafkaTopicListTreeNode) {
			List<TreeItem<KafkaTreeNode>> childrenList = new ArrayList<TreeItem<KafkaTreeNode>>();
			try {
				Collection<TopicListing> topics = kafkaClient.listTopics();
				for(TopicListing topic:topics) {
					KafkaTopicTreeNode topicTreeNode = new KafkaTopicTreeNode(topic.name());
					TreeItem<KafkaTreeNode> topicItem = new TreeItem<KafkaTreeNode>(topicTreeNode);
					childrenList.add(topicItem);
				}
			} catch (Exception e) {
				AlertUtil.exception(e);
			}
			return childrenList;
		}else if(selectedNode instanceof KafkaConsumerGroupListTreeNode) {
			List<TreeItem<KafkaTreeNode>> childrenList = new ArrayList<TreeItem<KafkaTreeNode>>();
			try {
				Collection<ConsumerGroupListing> consumerGroups = kafkaClient.listConsumerGroups();
				for(ConsumerGroupListing consumerGroup:consumerGroups) {
					KafkaConsumerGroupTreeNode consumerGroupTreeNode = new KafkaConsumerGroupTreeNode(consumerGroup.groupId());
					TreeItem<KafkaTreeNode> consumerGroupItem = new TreeItem<KafkaTreeNode>(consumerGroupTreeNode);
					childrenList.add(consumerGroupItem);
				}
			} catch (Exception e) {
				AlertUtil.exception(e);
			}
			return childrenList;
		}
		return null;
	}

	/**
	 * 获取节点信息
	 * @param selectedNode　当前选中节点
	 * @return　节点信息
	 */
	private String getItemInfo(KafkaTreeNode selectedNode) {
		appLogger.info("获取Kafka树节点详情信息。当前节点："+selectedNode);
		if(selectedNode instanceof KafkaRootTreeNode) {
			StringBuilder builder = new StringBuilder();
			DescribeClusterResult result = kafkaClient.describeCluster();
			try {
				String clusterId = result.clusterId().get();
				String controller = String.valueOf(result.controller().get());
				String nodes = String.valueOf(result.nodes().get());
				String authorizedOperations = String.valueOf(result.authorizedOperations().get());
				builder.append("clusterId:").append(clusterId).append("\n");
				builder.append("controller:").append(controller).append("\n");
				builder.append("nodes:").append(nodes).append("\n");
				builder.append("authorizedOperations:").append(authorizedOperations).append("\n");
			} catch (InterruptedException | ExecutionException e) {
				AlertUtil.exception(e);
			}
			return builder.toString();
		}else if(selectedNode instanceof KafkaTopicTreeNode) {
			String topic = ((KafkaTopicTreeNode) selectedNode).getTopic();
			TopicDescription topicInfo = kafkaClient.describeTopics(topic);
			return topicInfo.toString();
		}else if(selectedNode instanceof KafkaConsumerGroupTreeNode) {
			String groupId = ((KafkaConsumerGroupTreeNode) selectedNode).getGroupId();
			ConsumerGroupDescription consumerGroupInfo = kafkaClient.describeConsumerGroups(groupId);
			return consumerGroupInfo.toString();
		}
		return null;
	}

	/**
	 * 执行命令动作监听
	 * @param e　事件对象
	 */
	public void exeCmd(ActionEvent e) {
		if(kafkaClient == null) {
			appLogger.error("无法执行用户输入命令！【未建立连接】");
			return;
		}

		String cmdStr = cmdTextField.getText();
		if(cmdStr == null || "".equals(cmdStr.trim())) {
			appLogger.error("无法执行用户输入命令！【命令为空】");
			return;
		}

		appLogger.info("执行用户输入命令："+cmdStr);
		//执行Kafka命令
		String result = kafkaClient.exeCmd(cmdStr);
		//清空命令输入框
		cmdTextField.clear();
		//将命令和响应追加到命令执行历史内
		cmdTextArea.appendText(">"+cmdStr+"\n");
		cmdTextArea.appendText(result);
	}
}
